package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.common.GmallConstant;
import com.atguigu.realtime.udf.KeyWordUdtf;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.lang.reflect.Field;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/16 21:28
 */
public class DWSKeyWordStatsApp {
    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        setWebUi(env, 2000);
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env
            .getCheckpointConfig()
            .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop162:8020/gmall2021/flink/checkpoint2"));
        
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 1. 注册SourceTable: 从Kafka读数据
        // 如果是某个字段是json格式, 则把类型设置为map类型
        tenv.executeSql("CREATE TABLE page_view (" +
                            "   common MAP<STRING,STRING>, " +
                            "   page MAP<STRING,STRING>," +
                            "   ts BIGINT, " +
                            "   rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss'))," +
                            "   WATERMARK FOR  rowtime  AS  rowtime - INTERVAL '2' SECOND " +
                            ") WITH(" +
                            "   'connector' = 'kafka'," +
                            "   'topic' = 'dwd_page_log'," +
                            "   'properties.bootstrap.servers' = 'hadoop162:9029,hadoop163:9092,hadoop164:9092'," +
                            "   'properties.group.id' = 'DWSKeyWordStatsApp'," +
                            "   'scan.startup.mode' = 'latest-offset'," +
                            "   'format' = 'json'" +
                            ")");
        // 2. 注册SinkTable: 向ClickHouse写数据
        tenv.executeSql("create table keyword_stats_2021(" +
                            "   stt string," +
                            "   edt string," +
                            "   keyword string," +
                            "   source string," +
                            "   ct bigint," +
                            "   ts bigint," +
                            "   PRIMARY KEY (stt, edt, keyword, source) NOT ENFORCED" +
                            ")with(" +
                            "   'connector' = 'clickhouse', " +
                            "   'url' = 'clickhouse://hadoop162:8123', " +
                            "   'database-name' = 'gmall2021', " +
                            "   'table-name' = 'keyword_stats_2021'," +
                            "   'sink.batch-size' = '100', " +
                            "   'sink.flush-interval' = '1000', " +
                            "   'sink.max-retries' = '3' " +
                            ")");
        
        // 3. 从SourceTable查询数据, 并写入到SinkTable
        // 3.1 注册自定义函数  函数名, 函数类
        tenv.createTemporaryFunction("ik_analyze", KeyWordUdtf.class);
        // 3.2 过滤出需要的数据表
        final Table t1 = tenv.sqlQuery("select " +
                                           "    page['item'] fullword," +
                                           "    rowtime " +
                                           "from page_view " +
                                           "where page['item'] is not null " +
                                           "and page['page_id'] = 'good_list'");
        tenv.createTemporaryView("t1", t1);
        
        // 3.3 利用udtf进行拆分
        final Table t2 = tenv.sqlQuery("select " +
                                           "    keyword, " +
                                           "    rowtime " +
                                           "from t1, " +
                                           "lateral table(ik_analyze(fullword)) as T(keyword)");
        tenv.createTemporaryView("t2", t2);
        // 3.4 聚合
        final Table t3 = tenv.sqlQuery("select " +
                                           "    DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt, " +
                                           "    DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," +
                                           "    keyword, " +
                                           "    '" + GmallConstant.KEYWORD_SEARCH + "' source, " +
                                           "    count(*) ct, " +
                                           "    UNIX_TIMESTAMP()*1000 ts " +
                                           "from t2 " +
                                           "group by TUMBLE(rowtime, INTERVAL '10' SECOND ),keyword ");
        tenv.createTemporaryView("t3", t3);
        
        // 3.5 数据写入到ClickHouse
        tenv.executeSql("insert into keyword_stats_2021 select * from t3");
    }
    
    public static void setWebUi(StreamExecutionEnvironment env, int port) {
        try {
            final Field field = StreamExecutionEnvironment.class.getDeclaredField("configuration");
            field.setAccessible(true);
            final Configuration config = (Configuration) field.get(env);
            config.setInteger("rest.port", port);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
